package bx.job.actors;

import akka.actor.typed.ActorRef;
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.AbstractBehavior;
import akka.actor.typed.javadsl.ActorContext;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.Receive;
import akka.http.javadsl.Http;
import akka.http.javadsl.marshallers.jackson.Jackson;
import akka.http.javadsl.model.*;
import bx.InteractionEntity.CommonResponse;
import bx.InteractionEntity.JobWorkRequest;
import bx.job.JobServerBootstrap;
import bx.job.entity.BxJobInfo;
import bx.job.entity.BxJobWorkerInfo;
import bx.job.entity.BxLogInfo;
import bx.job.jobrouter.IJobWorkerRouter;
import bx.job.jobrouter.JobRouter;
import bx.job.jobstore.bean.JobInfoContextBean;
import com.google.gson.Gson;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;

import static bx.job.entity.BxLogInfo.*;

public class JobIssueActor extends AbstractBehavior<JobIssueActor.JobIssueActorMsg> {
    private final static Logger log = LoggerFactory.getLogger(JobIssueActor.class);
    private final static String EXECUTE_PATH = "/exec";

    public interface JobIssueActorMsg {
    }


    public record JobIssueMsg(JobInfoContextBean jobInfoContext) implements JobIssueActor.JobIssueActorMsg {
    }

    public record JobRestartIssueMsg(JobInfoContextBean jobInfoContext,
                                     BxJobWorkerInfo failedWorker,
                                     List<BxJobWorkerInfo> jobWorkerInfoList) implements JobIssueActor.JobIssueActorMsg {
    }

    public static Behavior<JobIssueActor.JobIssueActorMsg> create() {
        return Behaviors.setup(JobIssueActor::new);
    }

    private JobIssueActor(ActorContext<JobIssueActor.JobIssueActorMsg> context) {
        super(context);
    }

    @Override
    public Receive<JobIssueActor.JobIssueActorMsg> createReceive() {
        return newReceiveBuilder()
                .onMessage(JobIssueActor.JobIssueMsg.class, this::onJobIssueMsg)
                .onMessage(JobIssueActor.JobRestartIssueMsg.class, this::onJobRestartIssueMsg)
                .build();
    }

    // 重调度，先复用当前已经从数据库中查询到的 worker 列表，若为空，那么尝试从数据库获取新的 worker 列表重试


    // 正常调度
    private Behavior<JobIssueActorMsg> onJobIssueMsg(JobIssueMsg jobIssueMsg) {
        JobInfoContextBean jobInfoContext = jobIssueMsg.jobInfoContext();
        if (jobInfoContext == null) {
            log.error("JobInfoContext is null");
            return this;
        }
        CompletableFuture.supplyAsync(() ->
                        JobServerBootstrap.getJobWorkerStore().getJobWorkerListByWorkerGroupId(jobInfoContext.getBxJobInfo().getJobWorkerGroupId()))
                .thenAccept(jobWorkerInfoList -> {
                    if (jobWorkerInfoList.isEmpty()) {
                        log.error("current job worker list is empty . job info: {}", jobInfoContext.getBxJobInfo());
                        tellLogActorWithErrorCodeAndStatus(NO_WORKER_LIST_ERROR_CODE, FAILED_COMPLETED_STATUS, jobInfoContext.getLogInfo(), jobInfoContext.getLogActor());
                        return;
                    }
                    toIssueWorker(jobInfoContext, jobWorkerInfoList);
                });
        return this;
    }

    private Behavior<JobIssueActorMsg> onJobRestartIssueMsg(JobRestartIssueMsg jobRestartIssueMsg) {
        JobInfoContextBean jobInfoContext = jobRestartIssueMsg.jobInfoContext();
        BxJobInfo bxJobInfo = jobInfoContext.getBxJobInfo();
        Set<String> failedWorkerInfoSet = jobInfoContext.getFailedWorkerAddrInfo();
        BxJobWorkerInfo failedWorker = jobRestartIssueMsg.failedWorker();
        failedWorkerInfoSet.add(failedWorker.getIpAddressAndPort());
        if (bxJobInfo.getJobFailRetryCount() > failedWorkerInfoSet.size()) {
            log.info("current job limited retry count, job info :{}, failed worker info list :{}", bxJobInfo, failedWorkerInfoSet);
            tellLogActorWithErrorCodeAndStatus(RETRY_COUNT_LIMITED_ERROR_CODE, FAILED_COMPLETED_STATUS, jobInfoContext.getLogInfo(), jobInfoContext.getLogActor());
            return this;
        }
        jobRestartIssueMsg.jobWorkerInfoList.remove(failedWorker);
        if (jobRestartIssueMsg.jobWorkerInfoList.isEmpty()) {
            // 当前 worker list 已经全部重试完毕，那么从数据库中获取最新的 jobWorkerInfoList，若 jobWorkerInfoList 中的信息已经都存在于 failedWorkerInfoList，那么设置 log 调度失败
            CompletableFuture<List<BxJobWorkerInfo>> jobWorkerListCompletableFuture = CompletableFuture.supplyAsync(() ->
                    JobServerBootstrap.getJobWorkerStore().getJobWorkerListByWorkerGroupId(jobInfoContext.getBxJobInfo().getJobWorkerGroupId()));
            jobWorkerListCompletableFuture.thenAccept(workerInfoList -> {
                if (workerInfoList == null || workerInfoList.isEmpty()) {
                    log.error("current job's worker list is empty, job info :{}", bxJobInfo);
                    tellLogActorWithErrorCodeAndStatus(NO_AVAILABLE_WORKER_ERROR_CODE, FAILED_COMPLETED_STATUS, jobInfoContext.getLogInfo(), jobInfoContext.getLogActor());
                    return;
                }
                // 找到新注册的 worker 重试
                List<BxJobWorkerInfo> effectiveWorkerList = workerInfoList.stream().filter(workerInfo -> !failedWorkerInfoSet.contains(workerInfo.getIpAddressAndPort())).toList();
                if (effectiveWorkerList.isEmpty()) {
                    tellLogActorWithErrorCodeAndStatus(NO_AVAILABLE_WORKER_ERROR_CODE, FAILED_COMPLETED_STATUS, jobInfoContext.getLogInfo(), jobInfoContext.getLogActor());
                } else { // 继续尝试
                    tellLogActorIncrementRetryCount(jobInfoContext.getLogInfo(), jobInfoContext.getLogActor());
                    toIssueWorker(jobInfoContext, effectiveWorkerList);
                }
            });
        } else { // 移除后 继续尝试
            tellLogActorIncrementRetryCount(jobInfoContext.getLogInfo(), jobInfoContext.getLogActor());
            toIssueWorker(jobInfoContext, jobRestartIssueMsg.jobWorkerInfoList);
        }
        return this;
    }

    private void toIssueWorker(JobInfoContextBean jobInfoContext, List<BxJobWorkerInfo> jobWorkerInfoList) {
        IJobWorkerRouter.WorkerRouterContext workerRouterContext = IJobWorkerRouter.WorkerRouterContext.builder()
                .bxJobInfo(jobInfoContext.getBxJobInfo())
                .triggerInfo(jobInfoContext.getTriggerInfo())
                .workerInfoList(jobWorkerInfoList).build();
        IJobWorkerRouter iJobWorkerRouter = JobRouter.getWorkerRouterOrDefault(jobInfoContext.getBxJobInfo().getRouteStrategyName());
        List<IJobWorkerRouter.WorkerJobBindInfo> workerJobBindInfos = iJobWorkerRouter.jobRoute(workerRouterContext);
        workerJobBindInfos.forEach(bindInfo -> {
            String ipAddressAndPort = bindInfo.getBxJobWorkerInfo().getIpAddressAndPort();
            if (ipAddressAndPort == null) {
                log.error("current IP address info is null, bindInfo: {}", bindInfo);
                return;
            }
            CompletableFuture.runAsync(() -> {
                log.info("dispatch worker: {}", bindInfo);
                Gson gson = new Gson();
                JobWorkRequest jobWorkRequest = new JobWorkRequest();
                // todo 组装 请求信息
                HttpRequest httpRequest = HttpRequest.create(ipAddressAndPort + EXECUTE_PATH).withMethod(HttpMethods.POST)
                        .withEntity(HttpEntities.create(ContentTypes.APPLICATION_JSON, gson.toJson(jobWorkRequest)));
                final CompletionStage<HttpResponse> responseFuture = Http.get(this.getContext().getSystem()).singleRequest(httpRequest);
                responseFuture.whenComplete((response, error) -> {
                    if (error != null) {
                        log.error("dispatch worker failed, worker info :{}", bindInfo.getBxJobWorkerInfo(), error);
                        if (jobInfoContext.getBxJobInfo().getJobFailRetryCount() == 0) { // 不重试任务
                            log.info("current job no retry count, job info :{}", jobInfoContext.getBxJobInfo());
                            tellLogActorWithErrorCodeAndStatus(DISPATCH_WORKER_ERROR_CODE, FAILED_COMPLETED_STATUS, jobInfoContext.getLogInfo(), jobInfoContext.getLogActor());
                            return;
                        }
                        getContext().getSelf().tell(new JobRestartIssueMsg(jobInfoContext, bindInfo.getBxJobWorkerInfo(), jobWorkerInfoList));
                        return;
                    }
                    CompletionStage<CommonResponse> unmarshal = Jackson.unmarshaller(CommonResponse.class).unmarshal(response.entity(), this.getContext().getSystem());
                    unmarshal.toCompletableFuture().whenComplete((res, throwable) -> {
                        if (throwable == null && res.getStatus() == CommonResponse.SUCCESS_STATUS) {
                            log.info("dispatch worker success, worker info :{}", bindInfo.getBxJobWorkerInfo());
                            tellLogActorWithStatus(ISSUED_STATUS, jobInfoContext.getLogInfo(), jobInfoContext.getLogActor());
                        } else {
                            log.info("dispatch worker failed, worker info:{},response:{}", bindInfo.getBxJobWorkerInfo(), res, throwable);
                            if (jobInfoContext.getBxJobInfo().getJobFailRetryCount() == 0) { // 不重试任务
                                log.info("current job no retry count, job info :{}", jobInfoContext.getBxJobInfo());
                                tellLogActorWithErrorCodeAndStatus(DISPATCH_WORKER_ERROR_CODE, FAILED_COMPLETED_STATUS, jobInfoContext.getLogInfo(), jobInfoContext.getLogActor());
                                return;
                            }
                            getContext().getSelf().tell(new JobRestartIssueMsg(jobInfoContext, bindInfo.getBxJobWorkerInfo(), jobWorkerInfoList));
                        }
                    });
                });
            });
        });
    }


    private void tellLogActorWithErrorCodeAndStatus(byte errorCode, byte status, BxLogInfo logInfo, ActorRef<JobLogActor.JobIssueActorMsg> jobLogActor) {
        logInfo.setStatus(status);
        logInfo.setErrorCode(errorCode);
        JobLogActor.UpdateJobStatusWithErrorCodeMsg updateJobInfoMsg = new JobLogActor.UpdateJobStatusWithErrorCodeMsg(logInfo);
        jobLogActor.tell(updateJobInfoMsg);
    }

    private void tellLogActorWithStatus(byte status, BxLogInfo logInfo, ActorRef<JobLogActor.JobIssueActorMsg> jobLogActor) {
        tellLogActorWithErrorCodeAndStatus((byte) 0, status, logInfo, jobLogActor);
    }

    private void tellLogActorIncrementRetryCount(BxLogInfo logInfo, ActorRef<JobLogActor.JobIssueActorMsg> jobLogActor) {
        logInfo.setRetryCount((byte) (logInfo.getRetryCount() + 1));
        JobLogActor.UpdateJobRetryCount updateJobInfoMsg = new JobLogActor.UpdateJobRetryCount(logInfo);
        jobLogActor.tell(updateJobInfoMsg);
    }
}
